package ai.leinao.alarm.limit;

import ai.leinao.alarm.domain.AlarmGroupConverge;
import ai.leinao.alarm.domain.GroupAlarm;
import ai.leinao.alarm.domain.RawAlarm;
import ai.leinao.alarm.service.AlarmCroupConvergeService;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.thread.ThreadFactoryBuilder;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * 告警分组收敛
 *
 * @author zhong.wang
 * @date 2025/9/3
 **/
@Component
@Slf4j
public class AlarmGroupLimitProcessor {

    /**
     * 检查发送间隔
     */
    private static final long CHECK_INTERVAL = 1000L;


    /**
     * 默认初始分组等待时间 30秒
     */
    private static final long DEFAULT_GROUP_WAIT = 30 * 1000L;

    /**
     * 默认分组发送间隔 5分钟
     */
    private static final long DEFAULT_GROUP_INTERVAL = 5 * 60 * 1000L;

    /**
     * 告警抑制接口
     */
    private final AlarmInhibitProcessor alarmInhibitProcessor;


    /**
     * 告警分组缓存容器
     * key: groupKey
     * value: GroupAlertCache
     */
    private final Map<String, GroupAlarmCache> groupCacheMap;

    /**
     * 告警分组服务类
     */
    private final AlarmCroupConvergeService alarmGroupConvergeService;


    /**
     * 告警分组收敛规则容器
     * key: groupName
     * value: AlertGroupConverge
     */
    private final Map<String, AlarmGroupConverge> groupDefines;


    public AlarmGroupLimitProcessor(AlarmInhibitProcessor alarmInhibitProcessor, AlarmCroupConvergeService alarmGroupConvergeService) {
        this.alarmInhibitProcessor = alarmInhibitProcessor;
        this.alarmGroupConvergeService = alarmGroupConvergeService;
        this.groupCacheMap = new ConcurrentHashMap<>(8);
        this.groupDefines = new ConcurrentHashMap<>(8);
        initGroupDefines();
        startCheckAndSendGroups();
    }

    /**
     * 初始化告警分组收敛规则
     */
    private void initGroupDefines() {
        alarmGroupConvergeService.getAlarmGroupConverge().stream().forEach(group -> {
            groupDefines.put(group.getName(), group);
        });
    }


    /**
     * 定时检查并发送分组告警
     */
    private void startCheckAndSendGroups() {
        ThreadFactory threadFactory = new ThreadFactoryBuilder()
                .setUncaughtExceptionHandler((thread, throwable) -> {
                    log.error("检查告警分组计算时发生未捕获异常。");
                    log.error(throwable.getMessage(), throwable);
                })
                .setDaemon(true)
                .setNamePrefix("alarm-group-calculate-%d")
                .build();
        ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor(threadFactory);
        scheduledExecutor.scheduleAtFixedRate(() -> {
            try {
                long now = System.currentTimeMillis();
                groupCacheMap.forEach((groupKey, cache) -> {
                    if (shouldSendGroup(cache, now)) {
                        sendGroupAlarm(cache);
                        cache.setLastSendTime(now);
                        cache.getAlarmFingerprints().clear();
                    }
                });
            } catch (Exception e) {
                log.error("检查告警分组计算时发生异常: {}", e.getMessage(), e);
            }
        }, 1000, CHECK_INTERVAL, TimeUnit.MILLISECONDS);
    }
    

    /**
     * 告警分组收敛处理逻辑
     * @param alarm
     */
    public void processGroupAlert(RawAlarm alarm) {
        //log.info("告警分组收敛处理中...");
        Map<String,String> labels = alarm.getContext();
        // 告警内容为空，发送单条告警
        if(CollUtil.isEmpty(labels)){
            sendSingleAlert(alarm);
            return;
        }
        boolean matched = false;
        for (Map.Entry<String, AlarmGroupConverge> define : groupDefines.entrySet()) {
            String defineName = define.getKey();
            AlarmGroupConverge ruleConfig = define.getValue();

            // 匹配分组标签
            if (hasRequiredLabels(labels, ruleConfig.getGroupLabels())) {
                // 匹配成功，进行分组收敛处理
                matched = true;
                processAlertByGroupDefine(alarm, defineName, ruleConfig);
            }
        }

        // 如果存在没有匹配的规则，则发送单条告警
        if (!matched) {
            sendSingleAlert(alarm);
        }
    }

    /**
     * 根据分组定义处理告警
     * @param alarm
     * @param defineName
     * @param ruleConfig
     */
    private void processAlertByGroupDefine(RawAlarm alarm, String defineName, AlarmGroupConverge ruleConfig){
        // 基于定义提取分组标签
        Map<String, String> extractedLabels = new HashMap<>();
        Map<String,String> labels = alarm.getContext();
        for (String labelKey : ruleConfig.getGroupLabels()) {
            extractedLabels.put(labelKey, labels.get(labelKey));
        }

        // 生成分组键
        String groupKey = generateGroupKey(extractedLabels);

        // 获取或创建分组缓存
        GroupAlarmCache cache = groupCacheMap.computeIfAbsent(groupKey, k -> {
            GroupAlarmCache newCache = new GroupAlarmCache();
            newCache.setGroupKey(groupKey);
            newCache.setGroupLabels(extractedLabels);
            newCache.setGroupDefineName(defineName);
            newCache.setCreateTime(System.currentTimeMillis());
            newCache.setAlarmFingerprints(new ConcurrentHashMap<>(8));
            return newCache;
        });
        String fingerprint = alarm.getFingerprint();
        // 检查当前告警是否为重复
        RawAlarm existingAlarm = cache.getAlarmFingerprints().get(fingerprint);
        if (existingAlarm != null) {
            // 更新告警的开始时间，维持初次告警时间
            alarm.setAlarmStartTime(existingAlarm.getAlarmStartTime());
            alarm.setAlarmCount(existingAlarm.getAlarmCount() + 1);
            cache.getAlarmFingerprints().put(fingerprint, alarm);
            return;
        }

        // Add new alert
        cache.getAlarmFingerprints().put(fingerprint, alarm);
    }


    /**
     * 检查告警标签是否包含所有必需的分组标签
     *
     * @param labels         告警的标签映射
     * @param requiredLabels 必需的分组标签列表
     * @return 如果告警标签包含所有必需的分组标签则返回true，否则返回false
     */
    private boolean hasRequiredLabels(Map<String, String> labels, List<String> requiredLabels) {
        return requiredLabels.stream().allMatch(labels::containsKey);
    }

    /**
     * 发送单条告警
     * @param alarm
     */
    private void sendSingleAlert(RawAlarm alarm) {
        // Wrap single alert as group alert
        String groupKey = generateGroupKey(alarm.getContext());
        GroupAlarm groupAlert = GroupAlarm.builder()
                .groupKey(groupKey)
                .groupLabels(alarm.getContext())
                .alarms(Arrays.asList(alarm))
                .alarmFingerprints(Arrays.asList(alarm.getFingerprint()))
                .build();
        alarmInhibitProcessor.inhibitAlarm(groupAlert);
    }

    /**
     * 发送告警分组
     * @param cache
     */
    private void sendGroupAlarm(GroupAlarmCache cache) {
        if(CollUtil.isEmpty(cache.getAlarmFingerprints())){
            return;
        }
        GroupAlarm groupAlarm = GroupAlarm.builder()
                .groupKey(cache.getGroupKey())
                .groupLabels(cache.getGroupLabels())
                .alarms(new ArrayList<>(cache.getAlarmFingerprints().values()))
                .alarmFingerprints(new ArrayList<>(cache.getAlarmFingerprints().keySet()))
                .build();
        alarmInhibitProcessor.inhibitAlarm(groupAlarm);
    }


    /**
     * 判断是否应该发送告警分组
     * @param cache
     * @param now
     * @return
     */
    private boolean shouldSendGroup(GroupAlarmCache cache, long now) {
        AlarmGroupConverge ruleConfig = groupDefines.get(cache.getGroupDefineName());
        long groupWait = Objects.nonNull(ruleConfig) && Objects.nonNull(ruleConfig.getGroupWait()) ? ruleConfig.getGroupWait() * 1000 : DEFAULT_GROUP_WAIT;
        long groupInterval = Objects.nonNull(ruleConfig) && Objects.nonNull(ruleConfig.getGroupInterval())  ? ruleConfig.getGroupInterval() : DEFAULT_GROUP_INTERVAL;

        // First wait time reached
        if (cache.getLastSendTime() == 0
                && now - cache.getCreateTime() >= groupWait) {
            return true;
        }
        // Group interval time reached
        return cache.getLastSendTime() > 0
                && now - cache.getLastSendTime() >= groupInterval;
    }

    /**
     * 生成分组键
     * @param labels
     * @return
     */
    private String generateGroupKey(Map<String, String> labels) {
        return labels.entrySet().stream()
                .sorted(Map.Entry.comparingByKey())
                .map(e -> e.getKey() + ":" + e.getValue())
                .collect(Collectors.joining(","));
    }


    /**
     * 分组收敛缓存
     */
    @Data
    private static class GroupAlarmCache {
        /**
         * 分组收敛规则名称
         */
        private String groupDefineName;
        /**
         * 分组收敛key
         */
        private String groupKey;

        /**
         * 分组标签
         */
        private Map<String, String> groupLabels;

        /**
         * 告警列表
         */
        private Map<String, RawAlarm> alarmFingerprints = new ConcurrentHashMap<>(8);

        /**
         * 创建时间
         */
        private long createTime;

        /**
         * 最后发送时间
         */
        private long lastSendTime;
    }
}
